package cloud.health.framework.computing.gateway.domain;

import cloud.health.framework.computing.core.AbstractNettyServer;
import com.alibaba.fastjson.JSON;
import com.google.common.primitives.Bytes;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
import io.netty.util.internal.StringUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;

@Slf4j
public class Gateway extends AbstractNettyServer {

    private Integer port;

    public Gateway() {

    }

    public Gateway(int port) {
        this.port = port;
    }

    private static final ConcurrentHashMap<String, Proxy> SERVER = new ConcurrentHashMap<>();
    private static final ConcurrentHashMap<String, ChannelHandlerContext> COMMANDS = new ConcurrentHashMap<>();
    private static final CopyOnWriteArrayList<Proxy.ProxyChannelHandler> HANDLERS = new CopyOnWriteArrayList<>();

    private static final CopyOnWriteArrayList<Byte> bytes = new CopyOnWriteArrayList<>();

    public static void find(Proxy.ProxyChannelHandler handler) {
        String port = handler.getContext().channel().localAddress().toString();
        port = port.substring(port.lastIndexOf(":") + 1);
        String finalPort = port;
        Optional<ChannelHandlerContext> opt = COMMANDS.entrySet().parallelStream()
                .filter(it -> {
                    return it.getKey().equals(finalPort);
                }).map(it -> it.getValue())
                .findFirst();
        if (!opt.isPresent()) {
            String res = "HTTP/1.1 404 NOT FOUND";
            handler.getContext().writeAndFlush(Unpooled.wrappedBuffer(res.getBytes()));
            handler.getContext();
            return;
        }
        synchronized (opt.get()) {
            Map<String, String> data = new HashMap<>();
            data.put("port", finalPort);
            data.put("target", handler.getContext().channel().id().asShortText());
            String str = JSON.toJSONString(data) + "\n";
            opt.get().writeAndFlush(Unpooled.wrappedBuffer(str.getBytes()));
            HANDLERS.add(handler);
            while (Objects.isNull(handler.getProxy())) {
                try {
                    Thread.sleep(1L);
                } catch (InterruptedException e) {
                    log.error("InterruptedException -->", e);
                }
            }
        }
    }

    @Override
    protected int port() {
        return this.port;
    }

    @Override
    protected ChannelInitializer<SocketChannel> channelInitializer() {
        return new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                log.info("ch = {}", ch);
                GatewayChannelInboundHandler handler = new GatewayChannelInboundHandler();
                ch.pipeline().addLast(handler);
            }
        };
    }

    public static class GatewayChannelInboundHandler extends ChannelInboundHandlerAdapter {

        private ChannelHandlerContext proxy;

        @Override
        public void channelActive(ChannelHandlerContext ctx) {
            log.info("channel is active, ctx = {}", ctx);
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) {
            log.info("channel is inactive, ctx = {}", ctx);
            if (Objects.nonNull(proxy)) {
                proxy.close();
            }
            COMMANDS.entrySet().parallelStream().forEach(it -> {
                if (it.getValue().channel().id().asShortText().equals(ctx.channel().id().asShortText())) {
                    COMMANDS.remove(it);
                }
            });
            HANDLERS.removeIf(it -> it.getContext().channel().equals(ctx.channel().id().asShortText()));
            ctx.close();
        }

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) {
            ByteBuf buf = (ByteBuf) msg;
            try {
                String str = buf.toString(StandardCharsets.UTF_8);
                log.info("channel read, ctx = {}, size = {},  buf = \n{}", ctx, buf.readableBytes(), str);
                if (Objects.nonNull(proxy)) {
                    proxy.writeAndFlush(buf.copy());
                    return;
                }
                if (!bytes.isEmpty() || str.startsWith("<!--") || str.endsWith("-->")) {
                    byte[] b = new byte[buf.readableBytes()];
                    buf.readBytes(b);
                    bytes.addAll(Bytes.asList(b));
                    String s = Unpooled.wrappedBuffer(Bytes.toArray(bytes)).toString(StandardCharsets.UTF_8);
                    if (s.startsWith("<!--") && s.endsWith("-->")) {
                        s = s.replaceAll("<!--", "").replaceAll("-->", "").trim();
                        if (s.startsWith("[")) {
                            List<Map> list = JSON.parseArray(s, Map.class);
                            list.forEach(it -> {
                                String ps = it.get("remote_port").toString();
                                SERVER.entrySet().forEach(kv -> {
                                    if (ps.equals(kv.getKey())) {
                                        kv.getValue().stop();
                                        SERVER.remove(kv);
                                    }
                                });
                                COMMANDS.put(ps, ctx);
                                Integer port = Integer.valueOf(ps);
                                try {
                                    Proxy p = new Proxy(port);
                                    p.run();
                                    SERVER.put(port.toString(), p);
                                } catch (Exception e) {
                                    log.error("server run error -->", e);
                                }
                            });
                        } else {
                            log.info("s = {}", s);
                            Map<String, Object> m = JSON.parseObject(s, Map.class);
                            String target = m.getOrDefault("target", "").toString();
                            Optional<Proxy.ProxyChannelHandler> opt = HANDLERS.stream()
                                    .filter(it -> it.getContext().channel().id().asShortText().equals(target))
                                    .findFirst();
                            if (opt.isPresent()) {
                                opt.get().setProxy(ctx);
                                this.proxy = opt.get().getContext();
                            }
                        }
                        bytes.clear();
                    }
                }
            } finally {
                buf.release();
            }
        }

    }

}
